package com.zidongxiangxi.rabbit.starter.configuration.consumer;

import com.zidongxiangxi.rabbit.consumer.config.ConsumeRecordClearConfig;
import com.zidongxiangxi.rabbit.consumer.interceptor.IdempotentOperationsInterceptor;
import com.zidongxiangxi.rabbit.consumer.interceptor.SequenceOperationsInterceptor;
import com.zidongxiangxi.rabbit.consumer.job.ConsumeRecordClearThread;
import com.zidongxiangxi.rabbit.consumer.manager.ConsumeRecordManager;
import com.zidongxiangxi.rabbit.starter.constant.BeanNameConstants;
import com.zidongxiangxi.rabbit.starter.entity.ReliableRabbitProperties;
import com.zidongxiangxi.rabbit.starter.entity.consumer.ReliableRabbitConsumerIdempotent;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.condition.ConditionalOnSingleCandidate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.support.TransactionTemplate;

import javax.sql.DataSource;

/**
 * 幂等消费配置类
 *
 * @author chenxudong
 * @date 2020/11/08
 */
@Configuration
@ConditionalOnProperty(prefix = "reliable-rabbit.consumer.idempotent", name = "enabled", havingValue = "true")
public class ReliableRabbitConsumerIdempotentConfiguration {
	private final ReliableRabbitProperties properties;

	public ReliableRabbitConsumerIdempotentConfiguration(ReliableRabbitProperties properties) {
		this.properties = properties;
	}
	/**
	 * 消费记录manager
	 *
	 * @param jdbcTemplate jdbcTemplate
	 * @return 消费记录manager
	 */
	@Bean
	@ConditionalOnMissingBean(ConsumeRecordManager.class)
	public ConsumeRecordManager consumeRecordManager(JdbcTemplate jdbcTemplate) {
		return new ConsumeRecordManager(properties.getConsumer().getIdempotent().getTableName(), jdbcTemplate);
	}

	/**
	 * 定义幂等消费的拦截器
	 *
	 * @param consumeRecordManager 消费记录manager
	 * @param transactionTemplateObjectProvider spring事务template的提供者
	 * @return 幂等消费的拦截器
	 */
	@Bean(name = BeanNameConstants.INTERNAL_IDEMPOTENT_OPERATIONS_INTERCEPTOR)
	@ConditionalOnMissingBean(IdempotentOperationsInterceptor.class)
	public IdempotentOperationsInterceptor idempotentOperationsInterceptor(
			ConsumeRecordManager consumeRecordManager,
			ObjectProvider<TransactionTemplate> transactionTemplateObjectProvider
	) {
		TransactionTemplate transactionTemplate = null;
		if (properties.getConsumer().getIdempotent().isTransaction()) {
			transactionTemplate = transactionTemplateObjectProvider.getIfUnique();
		}
		return new IdempotentOperationsInterceptor(properties.getApplication(), consumeRecordManager, transactionTemplate);
	}

	/**
	 * 定义清理消费记录的线程
	 *
	 * @param consumeRecordManager 消费记录manager
	 * @return 清理消费记录的线程
	 */
	@Bean
	@ConditionalOnProperty(prefix = "reliable-rabbit.consumer.idempotent", name = "auto-clear", havingValue = "true")
	public ConsumeRecordClearThread consumeRecordClearThread(ConsumeRecordManager consumeRecordManager) {
		ReliableRabbitConsumerIdempotent idempotent = properties.getConsumer().getIdempotent();
		ConsumeRecordClearConfig config = new ConsumeRecordClearConfig();
		config.setRetentionPeriod(idempotent.getRetentionPeriod());
		config.setClearBatchSize(idempotent.getClearBatchSize());
		config.setWorkPeriod(idempotent.getWorkPeriod());
		return new ConsumeRecordClearThread(config, consumeRecordManager);
	}

	/**
	 * 事务配置
	 */
	@ConditionalOnMissingBean(TransactionTemplate.class)
	@ConditionalOnProperty(prefix = "reliable-rabbit.consumer.idempotent", name = "transaction", havingValue = "true")
	protected static class ReliableRabbitTransactionConfiguration {
		/**
		 * 定义事务管理器
		 *
		 * @param dataSource 数据源
		 * @return 事务管理器
		 */
		@Bean
		@ConditionalOnMissingBean(PlatformTransactionManager.class)
		@ConditionalOnSingleCandidate(DataSource.class)
		public DataSourceTransactionManager reliableRabbitTransactionManager(DataSource dataSource) {
			return new DataSourceTransactionManager(dataSource);
		}

		/**
		 * 定义事务template
		 *
		 * @param reliableRabbitTransactionManager 事务管理器
		 * @return 事务template
		 */
		@Bean
		@ConditionalOnMissingBean(TransactionTemplate.class)
		@ConditionalOnSingleCandidate(DataSourceTransactionManager.class)
		public TransactionTemplate reliableRabbitTransactionTemplate(DataSourceTransactionManager reliableRabbitTransactionManager) {
			return new TransactionTemplate(reliableRabbitTransactionManager);
		}
	}

	/**
	 * 順序消費配置
	 */
	@ConditionalOnProperty(prefix = "reliable-rabbit.consumer.idempotent.sequence", name = "enabled", havingValue = "true")
	protected static class ReliableRabbitConsumerSequenceConfiguration {
		private final ReliableRabbitProperties properties;

		public ReliableRabbitConsumerSequenceConfiguration(ReliableRabbitProperties properties) {
			this.properties = properties;
		}

		/**
		 * 定义顺序消费拦截器
		 *
		 * @param consumeRecordManager 消费记录manager
		 * @return 顺序消费拦截器
		 */
		@Bean(name = BeanNameConstants.INTERNAL_SEQUENCE_OPERATIONS_INTERCEPTOR)
		@ConditionalOnMissingBean(SequenceOperationsInterceptor.class)
		public SequenceOperationsInterceptor sequenceOperationsInterceptor(ConsumeRecordManager consumeRecordManager) {
			return new SequenceOperationsInterceptor(properties.getApplication(), consumeRecordManager,
					properties.getConsumer().getIdempotent().getSequence().getConsumeFailDelay());
		}

	}
}
